package com.jzl.canal_consumer.bean;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author ZhenWuWang
 */
public class CanalMessage
{
    private long id;
    private List<byte[]> entries = new ArrayList<>();
    private boolean raw = true;
    private List<ByteString> rawEntries = new ArrayList<>();

    public CanalMessage()
    {

    }

    public void setMessage(Message message)
    {
        this.id = message.getId();
        this.raw = message.isRaw();
        List<CanalEntry.Entry> entries = message.getEntries();
        if (entries != null && !entries.isEmpty())
        {
            this.entries = entries
                    .stream()
                    .map(CanalEntry.Entry::toByteArray)
                    .collect(Collectors.toList());
        }
        rawEntries = message.getRawEntries();
    }


    public Message getMessage()
    {
        List<CanalEntry.Entry> entries = new ArrayList<>();
        if (this.entries != null && !this.entries.isEmpty())
        {
            entries = this.entries
                    .stream()
                    .map(bytes ->
                    {
                        try
                        {
                            return CanalEntry.Entry.parseFrom(bytes);
                        } catch (InvalidProtocolBufferException e)
                        {
                            e.printStackTrace();
                            throw new IllegalArgumentException("反序列化错误：" + e.getMessage());
                        }
                    })
                    .collect(Collectors.toList());
        }
        Message message = new Message(id, entries);
        message.setRawEntries(rawEntries);
        message.setRaw(raw);
        return message;
    }

}
